package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * spark作业执行的特点：
 * 1、spark作业是由行动算子触发执行的，没有行动算子，整个作业不会执行
 * 2、每遇到一次行动算子，整个作业就会执行一次，有n个行动算子，作业就会执行n次，就会产生n个job
 *
 *
 * RDD: 弹性分布式数据集
 * 弹性：
 * RDD本身内部是不存储任何数据的，只存计算逻辑，当遇到行动算子的时候，触发作业执行，数据会由数据源读取进来
 * 读取的数据量可以是很大，也可以是很小。
 * 分布式：
 * 将来我们的spark作业是运行在hadoop集群中的，我们要读取的数据可能是hadoop中的block块，而block块可能是在不同的子节点上
 * 将来我们的计算任务可能在不同的节点上面执行，RDD的数据来自于不同的节点。
 * 数据集：
 * 当运行的时候，一个RDD可以被看作一个暂时存储数据的容器，在运行过程中，运行到这个RDD的时候，这一时刻的RDD的里面是有数据的。
 *
 *
 * 面试题：RDD的五大特性：
 * 1）RDD是有若干个分区组成
 *      a. 第一个RDD的分区，若没有手动指定的话，分区数等于读取的block块数【128*1.1】；若读取的是hdfs的数据，默认使用的是hadoop的读取方式
 *         b. 一开始读取数据源的算子中可以设置分区数
 *         若小于实际的block数量，由实际的数量来决定分区；
 *         若大于实际的block数量，取最小整数倍
 *         若后面的算子没有设置分区的话，一律由第一个RDD的分区数来决定。
 *         c. 数据源算子设置的分区数由上一个b结论得到最终的分区数，kv类算子可以单独设置分区数，会影响后面RDD的分区数
 *
 * 2） 算子是作用在每一个分区上的
 *      a. spark作业是由行动算子触发执行的，没有行动算子，整个作业不会执行
 *         b. 每遇到一次行动算子，整个作业就会执行一次，有n个行动算子，作业就会执行n次
 *
 * 3）RDD与RDD之间存在依赖关系
 *    a. 窄依赖  前一个RDD的分区数据，只会进入后一个RDD某一个分区，这两个RDD之间的关系叫做窄依赖。1对1的关系
 *       b. 宽依赖  前一个RDD的分区数据，会分散进入到后一个RDD中多个分区，这两个RDD之间的关系叫做宽依赖。1对n的关系
 *       c. 有宽依赖，会产生mr中类似逻辑shuffle，相同键到同一个分区中去
 *       d. 正是因为有了宽依赖，将整个spark作业划分若干个stage阶段。job(stage数量) = 宽依赖的数量 + 1
 *       e. 每个阶段的任务数量，取决于最大的分区数量
 *       f. 同一个spark进程中，若多个job作业具有相同的阶段执行，spark会进行一次优化，会将第一次执行的第一个阶段中的数据默认缓存
 *       在内存中，后续其他作业，可以跳过相同的阶段，直接从缓存中取出数据向后面RDD中传入执行。加快执行速度。
 *       后面我们会学习如何缓存到磁盘中。
 *
 * 4）kv类的算子，只能作用在kv类的RDD上
 *
 * 5） spark的作业执行过程中，只移动计算，不移动数据。
 *
 */

object WordCount2 {
  def main(args: Array[String]): Unit = {
    //1. 创建一个Spark配置文件对象
    val conf: SparkConf = new SparkConf()
    //可以对当前的spark作业设置一些值
    //spark程序可以运行在本地（local），或打包放在linux环境中运行【读写hdfs数据】（yarn）
    //如果只是在本地测试运行的话，需要将setMaster设置为local
    //如果将来程序在yarn上运行，setMaster不需要设置
    conf.setMaster("local")
    //给当前spark作业起一个名字
    conf.setAppName("spark入门程序单词统计本地运行")

    //2. 创建SparkContext对象,执行作业的核心对象，可以通过该对象获取spark作业的一些配置
    val sc: SparkContext = new SparkContext(conf)

    //3. 使用SparkContext读取本地文件数据【如果是本地运行，这里路径是本地路径；如果在linux环境中运行，这里路径是hdfs路径】
    //textFile函数，将文本数据中的每一行在了RDD中
    /**
     * RDD是spark core中的核心数据结构，不存储数据，存储计算逻辑，运行的时候，会有在其中数据流动
     * "spark/data/wc/"
     *
     * textFile底层是hadoop的TextInputFile文件读取数据的，默认一个block块一个切片，对应RDD一个分区
     *
     * minPartitions：设置的是最小分区数
     */
    val lineRDD: RDD[String] = sc.textFile("spark/data/wc/*")
    //    println(s"linesRDD的分区数量为：${lineRDD.getNumPartitions}")


    //4. 每一行数据，按照指定分隔符进行切分，进行扁平化
    val wordsRDD: RDD[String] = lineRDD.flatMap((e: String) => e.split(" "))

    val wordKVRDD: RDD[(String, Int)] = wordsRDD.map((_, 1))

    val resRDD: RDD[(String, Int)] = wordKVRDD.reduceByKey((v1: Int, v2: Int) => (v1 + v2))

    resRDD.foreach(println)


  }
}
